package pipeline;

import two.phase.termination.AbstractTerminationThread;
import two.phase.termination.TerminationToken;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;

/**
 * Created by Administrator on 2017/3/17.
 */
public class WorkerThreadPipeDecorator<IN,OUT> implements Pipe<IN,OUT> {

    protected final BlockingQueue<IN> workQueue;

    private final Set<AbstractTerminationThread> workerThreads = new HashSet<AbstractTerminationThread>();

    private final TerminationToken terminationToken = new TerminationToken();

    private final Pipe<IN,OUT> delegate;

    public WorkerThreadPipeDecorator(Pipe<IN, OUT> delegate,int workerCount) {
        this(new SynchronousQueue<IN>(),delegate,workerCount);
    }

    public WorkerThreadPipeDecorator(BlockingQueue<IN> workQueue, Pipe<IN, OUT> delegate,int workerCount) {
        if(workerCount <= 0){
            throw new IllegalArgumentException("workerCount should be positive..");
        }
        this.workQueue = workQueue;
        this.delegate = delegate;
        for(int i = 0 ;i < workerCount ;i++){
            workerThreads.add(new AbstractTerminationThread(terminationToken) {
                @Override
                protected void doRun() throws Exception {
                    try{
                        dispatch();
                    }finally {
                        terminationToken.reservations.decrementAndGet();
                    }

                }
            });
        }
    }

    protected void dispatch() throws InterruptedException{
        IN input = workQueue.take();
        delegate.process(input);
    }

    @Override
    public void init(PipeContext pipeContext) {
        delegate.init(pipeContext);
        for (AbstractTerminationThread thread : workerThreads){
            thread.start();
        }
    }

    @Override
    public void process(IN input) throws InterruptedException {
        workQueue.put(input);
        terminationToken.reservations.incrementAndGet();
    }

    @Override
    public void shutdown(long timeout, TimeUnit timeUnit) {
        for (AbstractTerminationThread thread : workerThreads){
            thread.terminate();
            try{
                thread.join(TimeUnit.MILLISECONDS.convert(timeout,timeUnit));
            }catch (InterruptedException e){
                // do nothing;
            }
        }
    }

    @Override
    public void setNextPipe(Pipe<?, ?> nextPipe) {
        delegate.setNextPipe(nextPipe);
    }
}
